{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": true,
    "pycharm": {
     "name": "#%%\n",
     "is_executing": false
    }
   },
   "outputs": [],
   "source": [
    "from pyflink.table.udf import udf, FunctionContext, ScalarFunction\n",
    "from pyflink.table import DataTypes, EnvironmentSettings, StreamTableEnvironment\n",
    "from pyflink.common import Row\n",
    "from pyflink.table import DataTypes\n",
    "from pyflink.table.udf import udtaf, TableAggregateFunction"
   ]
  },
  {
   "cell_type": "markdown",
   "source": [
    "* 创建UDTAF函数"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "outputs": [],
   "source": [
    "class Top2(TableAggregateFunction):\n",
    "\n",
    "    def create_accumulator(self):\n",
    "        # 存储当前最大的两个值\n",
    "        return [None, None]\n",
    "\n",
    "    def accumulate(self, accumulator, input_row):\n",
    "        if input_row[0] is not None:\n",
    "            # 新的输入值最大\n",
    "            if accumulator[0] is None or input_row[0] > accumulator[0]:\n",
    "                accumulator[1] = accumulator[0]\n",
    "                accumulator[0] = input_row[0]\n",
    "            # 新的输入值次大\n",
    "            elif accumulator[1] is None or input_row[0] > accumulator[1]:\n",
    "                accumulator[1] = input_row[0]\n",
    "\n",
    "    def emit_value(self, accumulator):\n",
    "        yield Row(accumulator[0])\n",
    "        if accumulator[1] is not None:\n",
    "            yield Row(accumulator[1])\n",
    "\n",
    "\n",
    "top2 = udtaf(f=Top2(),\n",
    "             result_type=DataTypes.ROW([DataTypes.FIELD(\"a\", DataTypes.BIGINT())]),\n",
    "             accumulator_type=DataTypes.ARRAY(DataTypes.BIGINT()))\n",
    "\n"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n",
     "is_executing": false
    }
   }
  },
  {
   "cell_type": "markdown",
   "source": [
    "*构建PyFlink1.13运行环境"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "outputs": [
    {
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mNameError\u001b[0m                                 Traceback (most recent call last)",
      "\u001b[1;32m<ipython-input-1-d67bd95b5a6e>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m\u001b[0m\n\u001b[1;32m----> 1\u001b[1;33m \u001b[0menv_settings\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mEnvironmentSettings\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mnew_instance\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0min_streaming_mode\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0muse_blink_planner\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mbuild\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m      2\u001b[0m \u001b[0mt_env\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mStreamTableEnvironment\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mcreate\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0menvironment_settings\u001b[0m\u001b[1;33m=\u001b[0m\u001b[0menv_settings\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      3\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      4\u001b[0m t = t_env.from_elements([(1, 'Hi', 'Hello'),\n\u001b[0;32m      5\u001b[0m                          \u001b[1;33m(\u001b[0m\u001b[1;36m3\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;34m'Hi'\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;34m'hi'\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m,\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;31mNameError\u001b[0m: name 'EnvironmentSettings' is not defined"
     ],
     "ename": "NameError",
     "evalue": "name 'EnvironmentSettings' is not defined",
     "output_type": "error"
    }
   ],
   "source": [
    "env_settings = EnvironmentSettings.new_instance().in_streaming_mode().use_blink_planner().build()\n",
    "t_env = StreamTableEnvironment.create(environment_settings=env_settings)\n",
    "\n",
    "t = t_env.from_elements([(1, 'Hi', 'Hello'),\n",
    "                         (3, 'Hi', 'hi'),\n",
    "                         (5, 'Hi2', 'hi'),\n",
    "                         (2, 'Hi', 'Hello'),\n",
    "                         (7, 'Hi', 'Hello')],\n",
    "                        ['a', 'b', 'c'])\n",
    "\n",
    "t_env.execute_sql(\"\"\"\n",
    "       CREATE TABLE my_sink (\n",
    "         word VARCHAR,\n",
    "         `sum` BIGINT\n",
    "       ) WITH (\n",
    "         'connector' = 'print'\n",
    "       )\n",
    "    \"\"\")\n",
    "\n",
    "result = t.group_by(t.b).flat_aggregate(top2).select(\"b, a\").execute_insert(\"my_sink\")\n",
    "\n",
    "result.wait()\n",
    "\n"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n",
     "is_executing": false
    }
   }
  }
 ],
 "metadata": {
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.6"
  },
  "kernelspec": {
   "name": "pyflink",
   "language": "python",
   "display_name": "pyflink"
  },
  "pycharm": {
   "stem_cell": {
    "cell_type": "raw",
    "source": [],
    "metadata": {
     "collapsed": false
    }
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}